package com.my.study.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

/**
 * @author Carlos
 * @version 1.0
 * @Description TODO
 * @date 2021/7/22 6:52
 **/
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    private KafkaTemplate<Object, Object> kafkaTemplate;

    @Autowired
    @SuppressWarnings("unchecked")
    public void setKafkaTemplate(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    //===============================topic01

    @GetMapping("/send01")
    public String sendTopic01(@RequestParam(value = "msg", required = false) String msg) {
        kafkaTemplate.send("topic01", msg);
        return "success";
    }

//    @KafkaListener(groupId="test", topics = "topic01")
    public void listener01(String msg) {
        System.out.println("==============Topic01==================");
        System.out.println("消费：" + msg);
    }

    //===============================topic02

    @GetMapping("/send02")
    public String sendTopic02(@RequestParam(value = "msg", required = false) String msg) {
        kafkaTemplate.send("topic02", msg);
        return "success";
    }

//    @KafkaListener(groupId="test", topics = "topic02")
    public void listener02(String msg) {
        System.out.println("==============Topic02==================");
        System.out.println("消费：" + msg);
    }

}
